package cn.fz.batch;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.nio.charset.StandardCharsets;
import java.util.*;

/**
 * 批量消息
 */
public class BatchProducer {
    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("BatchProducer");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        List<Message> messages = new ArrayList<>();
        for (int i = 0; i < 100000; i++) {
            Message message = new Message("Simple","Tags",(i+"_BatchProducer").getBytes(StandardCharsets.UTF_8));
            messages.add(message);
        }
        ListSplitter listsSplitter = new ListSplitter(messages);
        int i = 0;
        while (listsSplitter.hasNext()){
            SendResult sendResult = producer.send(listsSplitter.next());
            System.out.printf(i++ + "消息发送成功%s%n",sendResult);
        }
        producer.shutdown();
    }
}

class ListSplitter implements Iterator<List<Message>> {
    // private static final int SIZE_LIMIT = 1024 * 1024 * 4;
    // private static final int SIZE_LIMIT = 10 * 1000;// 发送限制1M
    private static final int SIZE_LIMIT = 40 * 1000;// 发送限制4M
    private final List<Message> messages;
    private int currentIndex;
    ListSplitter(List<Message> messages){
        this.messages = messages;
    }
    @Override
    public boolean hasNext() {
        return currentIndex < messages.size();
    }

    @Override
    public List<Message> next() {
        int nextIndex = currentIndex;
        int totalSize = 0;
        for (; nextIndex < messages.size(); nextIndex++) {
            Message message = messages.get(nextIndex);
            int messageSize = message.getBody().length + message.getTopic().length();
            Map<String, String> properties = message.getProperties();
            Iterator<Map.Entry<String, String>> iterator = properties.entrySet().iterator();
            while (iterator.hasNext()) {
                messageSize += iterator.next().getKey().length() + iterator.next().getValue().length();
            }
            messageSize = messageSize + 20; // 20是日志的大小
            if (messageSize > SIZE_LIMIT){
                // 第一次发送就超出了限制，直接跳过这条消息继续扫描
                if (nextIndex - currentIndex == 0){
                    nextIndex++;
                }
                break;
            }
            // 如果加上这条消息大于4M，则立即发送，否则继续扫描
            // 如果当前发送列表已经超出限制，暂停添加消息
            if (messageSize + totalSize > SIZE_LIMIT){
                break;
            }else{
                totalSize += messageSize;
            }
        }
        List<Message> subList = this.messages.subList(currentIndex, nextIndex);
        currentIndex = nextIndex;
        return subList;
    }
}